package com.fjeld.server.worker;

import com.fjeld.middleware.constants.PropertiesManager;
import com.fjeld.middleware.jaxb.V1Message;
import org.apache.log4j.Logger;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by IntelliJ IDEA.
 * User: fjeld
 * Date: 02/01/2012
 * Time: 23:21
 *
 * Dispatches items from one queue (usually from a listener transport) to
 * one of the @WorkerThread objects in it's array.
 *
 * Dispatch algorithm: Round robin of the WorkerThreads by using a LinkedList queue.
 * When a WorkerThread is needed, take it off the queue, assign the work, the put
 * the WorkerThread back on. As a WorkerThread has it's own work queue this should
 * be safe.
 */
public class Dispatcher implements Runnable
{
    private static final Logger LOGGER = Logger.getLogger(Dispatcher.class);
    private BlockingQueue<BlockingQueue> workerThreadPool;
    private BlockingQueue incomingMessageQueue;
    private String owner;

    public Dispatcher(String owner,
                      BlockingQueue incomingMessageQueue,
                      BlockingQueue<V1Message> publishQueue)
    {
        this.owner = owner;
        LOGGER.info("Creating dispatcher for " + owner);
        this.incomingMessageQueue = incomingMessageQueue;
        String queueSizeString = PropertiesManager.getProperties().getProperty("Dispatcher.WorkerThreadPoolSize", "1000");
        Integer queueSize = Integer.parseInt(queueSizeString);
        LOGGER.info("WorkerThreadPool: " + queueSize);

        workerThreadPool = new LinkedBlockingQueue<BlockingQueue>(queueSize);

        for (int i=0; i<queueSize; i++)
        {
            WorkerThread wt = new WorkerThread(i, publishQueue);
            workerThreadPool.add(wt.getWorkQueue());
            new Thread(wt).start();
        }

        LOGGER.info("Finished creation.");
    }

    public void run()
    {
        while(true)
        {
            try
            {
                LOGGER.info("Waiting for incoming message...");
                Object obj = incomingMessageQueue.take();

                LOGGER.info("Processing message...");
                BlockingQueue workerThreadQueue = workerThreadPool.take();
                workerThreadQueue.put(obj);
                workerThreadPool.put(workerThreadQueue);

                LOGGER.info("Finished processing message.");
            } catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
    }
}
